package com.example.event.rocket;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.EnableConfigurationProperties;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;

@EnableConfigurationProperties(EventProperties.class)
public class RocketMqProducer {

    private Logger logger = LoggerFactory.getLogger(getClass());
    @Resource
    private EventProperties eventProperties;

    private DefaultMQProducer producer;

    @Value(("${spring.application.name}"))
    private String appName;

    @PostConstruct
    private void defaultMqProducer() throws MQClientException {
        producer = new DefaultMQProducer(appName);
        producer.setNamesrvAddr(eventProperties.getNameServer());
        try {
            producer.start();
        } catch (MQClientException e) {
            logger.error("初始化生产都发生未知错误！", e);
            throw e;
        }
    }

    public void pushMessage(MqMessage message) throws Exception {
        String body = "";
        if (message.getBody() instanceof String) {
            body = (String) message.getBody();
        } else {
            body = JSON.toJSONString(message.getBody());
        }
        Message msg = new Message(message.getTopic(), message.getTag(), message.getKey(), body.getBytes());
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
            }

            @Override
            public void onException(Throwable throwable) {
                logger.error("发送消息发生知错误[{}]", JSON.toJSONString(message), throwable);
            }
        });
    }

    @PreDestroy
    public void shutdown() {
        producer.shutdown();
    }
}